﻿#include "mqttservice.h"
#include <algorithm>
#include <stdio.h>
#include <iostream>  
#include <sstream>  
#include <boost/format.hpp>
#include <boost/property_tree/ptree.hpp>
#include <boost/property_tree/json_parser.hpp>
#include <boost/regex.hpp>
#include <sys/time.h>
#include "../../liblog/Include/liblog.h"

const char* subtopic = "$sefon/v1/iot/things/";
const char* pubtopic = "$sefon/v1/iot/management/";
const int mq_qos = 1;
const int mq_keepalive = 60;

const char* msg_body = "body";
const char* msg_url = "url";
const char* action_get = "get";
const char* action_update = "update";
const char* action_ctrl = "update/documents";


using namespace boost::property_tree;
using namespace boost::gregorian;


CMQTTService::CMQTTService(const MQTTCONFIG config, IMqttEvent* event):
mosquittopp(config.strmqttid.c_str()),
m_mqttconfig(config),
m_event(event),
m_connect(false),
m_disonnect(false)
{
    m_strsubtopic = std::string(subtopic) + config.strmqttid;
    m_strpubtopic = std::string(pubtopic) + config.strmqttid;
    m_key[0] = 0;
    m_key[1] = 0;
    m_key[2] = 0;
    m_key[3] = 0;
    if (getDeviceKey(m_key, m_mqttconfig.strkey.c_str(), m_mqttconfig.strkey.length()))
    {
        return ;
    }
    
    //printf("key1:%x,key2:%x,key3:%x,key4:%x\n",m_key[0],m_key[1],m_key[2],m_key[3]);
    username_pw_set(config.strusrname.c_str(), config.strusrpwd.c_str());
    int ret = connect(config.strmqtthost.c_str(), config.mqttport, mq_keepalive);
    if (ret)
    {
        LogError(("%s:%s connect_async失败,返回错误码:%d\r\n", __PRETTY_FUNCTION__, config.strmqttid.c_str(), ret));
    }
    
    ret = loop_start();
    if (ret)
    {
        LogError(("%s:%s loop_start失败,返回错误码:%d\r\n", __PRETTY_FUNCTION__, config.strmqttid.c_str(), ret));
    }
    return;
}

CMQTTService::CMQTTService(IMqttEvent* event):
m_event(event)
,m_connect(false)
,m_disonnect(false)
{

}

int CMQTTService::Start(const MQTTCONFIG config)
{
    m_mqttconfig = config;
    m_strsubtopic = std::string(subtopic) + config.strmqttid;
    m_strpubtopic = std::string(pubtopic) + config.strmqttid;
    m_key[0] = 0;
    m_key[1] = 0;
    m_key[2] = 0;
    m_key[3] = 0;
    if (getDeviceKey(m_key, m_mqttconfig.strkey.c_str(), m_mqttconfig.strkey.length()))
    {
        return -1;
    }
    reinitialise(config.strmqttid.c_str(), true);
    username_pw_set(config.strusrname.c_str(), config.strusrpwd.c_str());
    
    int ret = connect(config.strmqtthost.c_str(), config.mqttport, mq_keepalive);
    if (ret)
    {
        LogError(("%s:%s connect_async失败,返回错误码:%d\r\n", __PRETTY_FUNCTION__, config.strmqttid.c_str(), ret));
    }
    
    ret = loop_start();
    //ret = loop(2, 1024);
    //ret = loop_misc();
    if (ret)
    {
        LogError(("%s:%s loop_start失败,返回错误码:%d\r\n", __PRETTY_FUNCTION__, config.strmqttid.c_str(), ret));
    }
    m_connect = true;
    return ret;
}

CMQTTService::~CMQTTService()
{
    unsubscribe(NULL, m_strsubtopic.c_str());
    Stop();
    loop_stop(true);
    LogDebug(("mqttid-%s 销毁\r\n", m_mqttconfig.strmqttid.c_str()));
}

void CMQTTService::Stop()
{
    if (!m_disonnect)
    {
        disconnect();
    }
}

int CMQTTService::disconnect()
{
    m_disonnect = true;
    return mosqpp::mosquittopp::disconnect();
}

int CMQTTService::reconnect()
{
    if (!m_disonnect)//被动断开才重连
         return mosqpp::mosquittopp::reconnect();
    return 0;
}
int CMQTTService::reconnect_async()
{
    if (!m_disonnect)//被动断开才重连
        return mosqpp::mosquittopp::reconnect_async();
    return 0;
}

static void showhex(const char* src, int len)
{
    if (!src)
        return;
    int i = 0;  
    for (;i < len; i++) 
    {
        printf("%02X ", (unsigned char)src[i]); 
    }
    printf("\n");
    return;
}

int CMQTTService::devSubscribe()
{
    int qos = mq_qos;
    return subscribe(NULL, m_strsubtopic.c_str(), qos);
}

int CMQTTService::pushUpdateData(int payloadlen, const void *payload)
{
    /*
    int len = XXTEAGetBufferLength(payloadlen, true);
    char * encprybuf = new char[len];
    memset(encprybuf, 0, len);
    XXTEAEncode(encprybuf, (const char*)payload, payloadlen, m_key);
    //printf("len = %d\n",len);
    //showhex(encprybuf,len);
    int ret = mosqpp::mosquittopp::publish(mid, topic, len, encprybuf, 1, 0);
    //int ret = mosqpp::mosquittopp::publish(mid, topic, payloadlen, payload, qos, retain);
    delete[] encprybuf;
    
    */
    std::string strpayload((const char*)payload, payloadlen);
    strpayload = getUpdateJson(strpayload);
    int qos = mq_qos;
    int ret = publish(NULL, m_strpubtopic.c_str(), strpayload.length(), strpayload.c_str(), qos, false);
    return ret;
}

int CMQTTService::pushGetData()
{
    std::string strpayload = getGetJson();
    int qos = mq_qos;
    int ret = publish(NULL, m_strpubtopic.c_str(), strpayload.length(), strpayload.c_str(), qos, false);
    return ret;
}

void CMQTTService::on_connect(int rc)
{
    if (rc == 0)
        m_connect = true;
    m_event->OnConnect(this, m_mqttconfig.strmqttid, rc);
}

void CMQTTService::on_disconnect(int rc)
{
    m_connect = false;
    if (rc == 0)
        m_event->OnDisconnect(this, m_mqttconfig.strmqttid, rc);
}

void CMQTTService::on_message(const struct mosquitto_message *message)
{
    //解密
    /*
    int len = XXTEAGetBufferLength(message->payloadlen, false);
    char * decbuf = new char[len];
    memset(decbuf, 0, len);   
    XXTEA_TYPE rlen = XXTEADecode(decbuf, (const char*)message->payload, message->payloadlen, m_key);
    if (rlen > len)
        rlen = len;
    */
    
    if (!message)
        return;

    //获取类型
    std::stringstream stream(std::string((const char*)message->payload, message->payloadlen));
    ptree pt;
    
    std::string strurl = "";
    std::string strbody = "";
    try{  
        read_json<ptree>(stream, pt);
        strurl = pt.get<std::string>(msg_url);
        std::stringstream bodystream;
        bodystream.clear();
        write_json(bodystream, pt.get_child(msg_body));
        strbody = bodystream.str();
    }catch(ptree_error & e) { 
        strurl = "";
        strbody = "";
        LogError(("%s: %s json form error\r\n", __PRETTY_FUNCTION__, (const char*)message->payload));
        return;   
    }

	const int matche_flags[] = { 1, 2}; // devid,actions
    boost::regex rule("^/things/([a-zA-Z0-9]{20})/shadow/(get|update|update/documents)$");
    boost::sregex_token_iterator it(strurl.begin(), strurl.end(), rule, matche_flags), end;
    if (it == end)
        return;
    std::string strmqid = *(it++);
    std::string straction = *it;

    if (strmqid.compare(m_mqttconfig.strmqttid) != 0)
    {
        LogError(("%s:数据中设备id(%s)与本设备id(%s)不匹配\r\n", __PRETTY_FUNCTION__,  strmqid.c_str(), m_mqttconfig.strmqttid.c_str()));
        return;
    }
    
    m_event->OnMessage(this, m_mqttconfig.strmqttid, GetActions(straction), strbody);
}

void CMQTTService::on_subscribe(int mid, int qos_count, const int *granted_qos)
{
     m_event->OnSubscribe(this, m_mqttconfig.strmqttid, mid, qos_count, granted_qos);
}

void CMQTTService::on_publish(int mid)
{
    m_event->OnPublish(this, m_mqttconfig.strmqttid,mid);
}

void CMQTTService::on_unsubscribe(int mid)
{
    m_event->OnPublish(this, m_mqttconfig.strmqttid,mid);
}


int CMQTTService::addJsonHeader(std::string & strdata, std::string strdevid ,ShadowActions action, std::string requestid)
{
    std::string strid = "";
    if (requestid.length()== 0)
        strid = getRequestId();
    else
        strid = requestid; 

    std::string strheader = "";
    switch(action)
    {
    case SHADOW_GET:
    {
        strheader = boost::str(boost::format("{\"requestId\":\"%s\",\"url\":\"/things/%s/shadow/get\"}")% strid % strdevid);
        break;
    }
    case SHADOW_UPDATE:
    {
        strheader = boost::str(boost::format("{\"requestId\":\"%s\",\"url\":\"/things/%s/shadow/update\","
                            "\"body\":{\"state\":{\"reported\":") %strid %strdevid);
        break;
    }
    case SHADOW_CTRL:
    {
        strheader =  boost::str(boost::format("{\"requestId\":\"%s\",\"url\":\"/things/%s/shadow/%s/%s\","
                        "\"body\":{\"state\":{\"reported\":{")%strid %strdevid % "update" %  "documents");

        break;
    }
    default:
        return -1;
    }

    std::string strpayload = strheader + strdata;
    strdata = strpayload;
    return 0;
}

void CMQTTService::addJsonTail(std::string & data, int version)
{
    if (version == 0)
        data += std::string("}}}");
    else
    {
        data += boost::str(boost::format("},\"version\":%d}}")%version);
    }
}

std::string CMQTTService::getRequestId(int strlen)
{
    struct timeval tv = {0, 0};
    uint32_t timeMs;
    static char chr[3] = {'0','a','A'};
    gettimeofday(&tv, NULL);
    timeMs = tv.tv_sec * 1000 + tv.tv_usec / 1000;
    srand(timeMs);
    int r = 0;
    std::string strid = "";
    int i = 0;
    for (; i < strlen; i++)
    {
        r = rand();
        if (0 ==(r % 3))
            strid += chr[r % 3] + rand() % 10;
        else
            strid += chr[r % 3] + rand() % 26;
    }
    
    return strid;
}

std::string CMQTTService::getUpdateJson(std::string & data, int version, std::string requestid)
{
    addJsonHeader(data, m_mqttconfig.strmqttid, SHADOW_UPDATE, requestid);
    addJsonTail(data, version);
    std::string strjson = data;
    data = strjson;
    return strjson;
}

std::string CMQTTService::getGetJson(std::string requestid)
{
    std::string data = "";
    addJsonHeader(data, m_mqttconfig.strmqttid, SHADOW_GET, requestid);
    std::string strjson = data;
    return strjson;
}

ShadowActions CMQTTService::GetActions(std::string straction)
{
    if (straction.compare(std::string(action_get))== 0)
        return SHADOW_GET;
    else if (straction.compare(std::string(action_update))== 0)
        return SHADOW_UPDATE;
    else
        return SHADOW_CTRL;
}



